package com.ganbo.rabbitmq.consumer.listener;

import com.ganbo.rabbitmq.common.enums.MessageStatus;
import com.ganbo.rabbitmq.consumer.config.DirectConfig;
import com.ganbo.rabbitmq.consumer.entity.ConsumerMessage;
import com.ganbo.rabbitmq.consumer.repository.ConsumerMessageRepository;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.util.Date;
import java.util.Map;

/**
 * 基本消息消费
 *
 * @Author: ganbo
 * @Date: 2020/12/8 19:21
 */


@Component
@RabbitListener(queues = DirectConfig.TOPIC_direct1) //配置监听的队列
public class BasicConsumer1 {

    @Autowired
    ConsumerMessageRepository consumerMessageRepository;
    @Autowired
    RestTemplate restTemplate;

    @RabbitHandler
    public void process(String msg, Channel channel, Message message) throws IOException {

        String messageId = message.getMessageProperties().getHeader("spring_returned_message_correlation");

        //接收到消息先本地保存
        ConsumerMessage consumerMessage = new ConsumerMessage();
        consumerMessage.setContent(msg);
        Date now = new Date();
        consumerMessage.setCreateTime(now);
        consumerMessage.setUpdateTime(now);
        consumerMessage.setMessageId(messageId);
        consumerMessage.setStatus(MessageStatus.e3.getCode());
        consumerMessageRepository.saveAndFlush(consumerMessage);

        //ack
        try {
            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("receiver success");


            //回调生产者
            ResponseEntity<String> forEntity = restTemplate.getForEntity("http://localhost:8090/callBack?messageId=" + messageId, String.class);
            System.out.println("消费者回调返回:====>" + forEntity.getBody());
        } catch (IOException e) {
            e.printStackTrace();
            //丢弃这条消息
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            System.out.println("receiver fail");
        }


        System.out.println("==========111==>消费者接受到消息=====>" + msg);
    }

}
